package com.vividstores.integration.mq.impl;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

import javax.xml.bind.JAXBException;

import org.apache.commons.io.FileUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.vividstores.integration.mq.MQFile;
import com.vividstores.integration.mq.MQFileBundle;
import com.vividstores.integration.mq.MQFileBundleHeader;
import com.vividstores.integration.mq.MQFileSender;

public class MQFileSenderImpl extends AbstractMQFileDispatcher implements
		MQFileSender {

	private static final int BUFFER_SIZE = 1024 * 128; // 128KB
	
	private MQQueueManager queueManager;
	private MQQueue outputQueue;

	@Override
	public void send(MQFileBundle fileBundle) throws FileNotFoundException,
			IOException, MQException {
		
		checkFilesValidity(fileBundle);
		
		initializeQueue();

		populateMetadata(fileBundle);

		MQPutMessageOptions pmo = new MQPutMessageOptions();
		pmo.options = MQC.MQPMO_LOGICAL_ORDER + MQC.MQPMO_SYNCPOINT;

		try {
			// create the control message in first message group.
			sendControlMessage(fileBundle, pmo);

			// create and segement the file content in remaining message groups.
			sendFileContent(fileBundle, pmo);

			queueManager.commit();
		} catch (MQException e) {
			queueManager.backout();
			throw e;
		} catch (FileNotFoundException e) {
			queueManager.backout();
			throw e;
		} catch (IOException e) {
			queueManager.backout();
			throw e;
		} catch (JAXBException e) {
			queueManager.backout();
			throw new RuntimeException(
					"Failed to marshal using JAXB, please check JAXB environment and model mapping.",
					e);
		} finally {
			if (outputQueue != null) {
				outputQueue.close();
			}
			queueManager.disconnect();
			queueManager.close();
		}
	}

	private void sendControlMessage(MQFileBundle fileBundle,
			MQPutMessageOptions pmo) throws IOException, JAXBException,
			MQException {
		MQMessage controlMessage = new MQMessage();
		controlMessage.characterSet = UTF_8_CHARSET_CODE;
		controlMessage.format = MQC.MQFMT_STRING;
		controlMessage.messageType = MQC.MQMT_DATAGRAM;
		// put this message in first group, this group has only one segement
		controlMessage.messageFlags = MQC.MQMF_MSG_IN_GROUP + MQC.MQMF_SEGMENT
				+ MQC.MQMF_LAST_SEGMENT;

		if (fileBundle.getFiles().size() == 0) {
			controlMessage.messageFlags += MQC.MQMF_LAST_MSG_IN_GROUP;
		}

		String marshallString = "";
		try {
			// XML serialization
			marshallString = marshall(fileBundle);
			logger.info("Sent file control info: " + marshallString);
		} catch (JAXBException e) {
			logger.error("Failed to marshal file info", e);
			throw e;
		} catch (UnsupportedEncodingException e) {
			logger.error("Charset " + UTF_8_CHARSET_NAME + " is not supported.", e);
			throw e;
		}

		try {
			controlMessage.writeString(marshallString);
		} catch (IOException e) {
			logger.error("Failed to string to control message.", e);
			throw e;
		}

		try {
			outputQueue.put(controlMessage, pmo);
		} catch (MQException e) {
			logger.error("Failed to put control message to queue.", e);
			throw e;
		}
	}

	@SuppressWarnings("unchecked")
	private void initializeQueue() throws MQException {
		MQEnvironment.hostname = hostname;
		MQEnvironment.port = port;
		MQEnvironment.channel = channel;
		MQEnvironment.CCSID = ccsid;
		MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,
				MQC.TRANSPORT_MQSERIES);

		queueManager = new MQQueueManager(queueManagerName);
		outputQueue = queueManager.accessQueue(queueName, MQC.MQOO_OUTPUT
				| MQC.MQOO_FAIL_IF_QUIESCING);
	}

	private void populateMetadata(MQFileBundle fileBundle) {
		Calendar calendar = Calendar.getInstance();
		Date currentTime = calendar.getTime();

		MQFileBundleHeader header = fileBundle.getHeader();
		header.setSendTime(currentTime);

		List<MQFile> files = fileBundle.getFiles();
		for (MQFile file : files) {
			file.setName(file.getFile().getName());
			file.setSize(file.getFile().length());
			calendar.setTimeInMillis(file.getFile().lastModified());
			file.setLastModified(calendar.getTime());
		}
	}

	private void checkFilesValidity(MQFileBundle fileBundle)
			throws FileNotFoundException, IOException {
		List<MQFile> files = fileBundle.getFiles();
		// test all files for reading
		for (int i = 0; i < files.size(); i++) {
			File file = files.get(i).getFile();
			FileUtils.openInputStream(file).close();
		}
	}

	private void sendFileContent(MQFileBundle fileBundle,
			MQPutMessageOptions pmo) throws FileNotFoundException, IOException,
			MQException {
		// when all files passed testing
		byte[] fileContent = new byte[BUFFER_SIZE];
		List<MQFile> files = fileBundle.getFiles();
		for (int i = 0; i < files.size(); i++) {
			MQFile mqFile = files.get(i);

			// this call should always success based on previous tests, not
			// exceptions thrown
			FileInputStream fis = null;

			try {
				fis = FileUtils.openInputStream(mqFile.getFile());
				MQMessage fileMessage = new MQMessage();
				fileMessage.format = MQC.MQFMT_NONE;
				fileMessage.messageType = MQC.MQMT_DATAGRAM;
				// segment the file content and put into the remaining group
				fileMessage.messageFlags = MQC.MQMF_MSG_IN_GROUP | MQC.MQMF_SEGMENT;

				if (i == (files.size() - 1)) {
					fileMessage.messageFlags += MQC.MQMF_LAST_MSG_IN_GROUP;
				}
				
				while (true) {
					fileMessage.clearMessage();
					int bytesRead = fis.read(fileContent, 0, BUFFER_SIZE);
					if (bytesRead == -1) {
						break;
					}
					fileMessage.write(fileContent, 0, bytesRead);
					if (fis.available() <= 0) {
						// reach the end of file, this is the last segment
						fileMessage.messageFlags += MQC.MQMF_LAST_SEGMENT;
					}
					outputQueue.put(fileMessage, pmo);
				}
			} finally {
				if (fis != null) {
					fis.close();
				}
			}
		}
	}
}
